package bidserver.src;

import java.net.*;
import java.text.ParseException;
import java.util.*;
import java.io.*;

enum Mode{
	TcpPeerClient, 
	TcpPeerServer,
	TcpClientServer
}
class ServerInfo {
	String name;
	int port;
	Socket sock = new Socket(); // null indicate not connected.

	public ServerInfo(String name, int port) {
		this.name = name;
		this.port = port;
	}
}

class TcpAcceptThread extends Thread {
	int port;
	Mode mode;
	BidServer server;

	TcpAcceptThread(int port, Mode mode, BidServer server) {
		this.port = port;
		this.mode = mode;
		this.server = server;
	}

	public void run() {
		ServerSocket sock = null;

		try {
			sock = new ServerSocket(port);
			Util.println("Tcp server to serve " + (mode == Mode.TcpClientServer?"clients":"peers")+" starts on " + port);
			while (true) {
				Socket peerSocket = sock.accept();
				TcpThread thread = new TcpThread(peerSocket, mode, server);
				thread.start();
			}
		} catch (IOException e) {
			Util.println("AcceptThread aborted due to IOException " + e);
		}
		if( sock != null )
			try {
				sock.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	}
}

class TcpThread extends Thread {
	Socket sock;
	Mode mode;
	BidServer server;

	public TcpThread(Socket sock, Mode mode, BidServer server) {
		this.sock = sock;
		this.mode = mode;
		this.server = server;
	}

	public void run() {
		PrintWriter pout = null;
		BufferedReader din = null;
		try {
			din = new BufferedReader(new InputStreamReader(sock.getInputStream()));
			pout = new PrintWriter(sock.getOutputStream());
			server.onConnectDisconnect(pout, mode, true);
			while (true) {
				String line = din.readLine();
				if( line == null)
					break;
				Msg msg = new Msg();
				msg.fromString(line);
				server.onMsg(msg, pout, mode);
			}
		} catch (IOException e) {
		} finally{
			try {
				sock.close();
				server.onConnectDisconnect(pout, mode, false);
				Util.println(mode == Mode.TcpClientServer ? "client disconnected." : "server disconnected.");
			} catch (IOException e) {
				
			}
		}
	}
}

public class BidServer extends ItemStore {
	int serverPort;// to accept server connections.
	int clientPort;// to accept client connection.
	List<PrintWriter> peers = new LinkedList<PrintWriter>();// used to conect to them. always larger id connect to smaller id.
	List<ServerInfo> smallerPeers = new LinkedList<ServerInfo>();
	RAReadWriteMutex lock = new RAReadWriteMutex(peers);
	volatile boolean initialItemBidsInProgress = false; // request sent to peer
 
	// server.

	public boolean init(int serverID, String conf) {
		if (!LoadConfig(serverID, conf))
			return false;

		// Start acceptThread to accept connections from Server
		TcpAcceptThread acceptServerThread = new TcpAcceptThread(serverPort, Mode.TcpPeerServer, this);
		acceptServerThread.start();
		// Now connect smaller peers.
		ConnectSmallerPeers();
		// Wait for some more for handshaking done.
		try {
			Thread.sleep(Constants.INITIALIZATION_TIMEOUT);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		// Now assume all necessary server-server connections are eastablished
		// and initial Hello msg exchanged.
		// Start to Sync myself to one of the peer.
		if (!peers.isEmpty()) {
			// assume I have items IDs in store.
			SyncItemStore(peers.get(0));
		}
		// Done initialization. start accepting clients.
		TcpAcceptThread acceptClientThread = new TcpAcceptThread(clientPort, Mode.TcpClientServer, this);
		acceptClientThread.start();
		return true;
	}

	public boolean LoadConfig(int serverId, String conf) {
		lock.srcId = serverId;
			Util.println(conf);
		try {
			BufferedReader din = new BufferedReader(new FileReader(conf));
			while (true) {
				String line = din.readLine();
                if (line == null) {
                    break;
                }
				line = line.trim();
				if (line.isEmpty())
					continue; // ignore empty.
				if (line.startsWith("#"))
					continue;// comment.

				// ID Name/IP ServerPort ClientPort
				String s[] = line.split("\\s+");
				if (s.length < 4)
					return false;
				int id = Integer.parseInt(s[0]);
				String serverName = s[1];
				int sPort = Integer.parseInt(s[2]);
				int cPort = Integer.parseInt(s[3]);
				if (id < serverId) {
					smallerPeers.add(new ServerInfo(serverName, sPort));
				} else if (id == serverId) {
					serverPort = sPort;
					clientPort = cPort;
				} else {
					break; // stop parsing.
				}
			}
		} catch (FileNotFoundException e) {
			return false;
		} catch (IOException e) {
			return false;
		}
		return true;
	}

	public void ConnectSmallerPeers() {
		for (ServerInfo server : smallerPeers) {
			if (!server.sock.isClosed() && server.sock.isConnected())
				continue;
			try {
				InetAddress addr;
				addr = InetAddress.getByName(server.name);
				InetSocketAddress destAddr = new InetSocketAddress(addr, server.port);
				Util.println("connecting to -> " + server.name + ":" + server.port);
				server.sock = new Socket();//new socket.
				server.sock.connect(destAddr, Constants.CONNECTION_TIMEOUT);
				Util.println("connected to peer " + server.name + ":" + server.port);
				TcpThread thread = new TcpThread(server.sock, Mode.TcpPeerClient, this);
				thread.start();
			} catch (SocketTimeoutException e) {
				try {
					server.sock.close();
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
				Util.println("Connecting to " + server.name + " " + server.port + "timeouted.");
			} catch (UnknownHostException e) {
				try {
					server.sock.close();
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
				Util.println(server.name + " is an unknown host");				
			} catch (IOException e) {
				try {
					server.sock.close();
				} catch (IOException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
				Util.println("Connecting to " + server.name + " " + server.port + " failed.");
			}
		}
	}

	public void HouseKeep() {// connecting to peer servers with lower IDs as tcp
		// client..
		while (true) {
			ConnectSmallerPeers();
			try {
				Thread.sleep(Constants.INITIALIZATION_TIMEOUT);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public void  SyncItemStore(PrintWriter pout) {
		for (int itemId : items.keySet()) {
			Msg msg = new Msg();
			msg.add("Request", "ItemBids");
			msg.add("TimeStamp", lock.c.tick());
			msg.add("ItemId", itemId);

			lock.requestCS(itemId, true);
			initialItemBidsInProgress = true;
			Util.sendMsg(msg, pout);
			while (initialItemBidsInProgress)
				Util.wait(this);
			lock.releaseCS(itemId);
		}
	}

	public void onMsg(Msg msg, PrintWriter pout, Mode mode) {
		String s = msg.toString();
		Util.println("OnMsg(" + s + ")");
		if (mode == Mode.TcpClientServer)
			onClientMsg(msg, pout);
		else
			onServerMsg(msg, pout);
	}

	public synchronized void onConnectDisconnect(PrintWriter pout, Mode mode, boolean connect) {
		if( mode != Mode.TcpClientServer ){
			if( connect ){
				if (mode == Mode.TcpPeerServer) {
					synchronized(peers){
						peers.add(pout);
					}
					Hello(pout);
				} else if( mode == Mode.TcpPeerClient){
					synchronized(peers){
						peers.add(pout);
					}
				}
			}
			else{
				if (pout != null) {
					synchronized(peers){
						peers.remove(pout);
					}
					//remove pouts in lock.
					lock.onDisconnect(pout);
				}
			}
		}
	}

	public void onClientMsg(Msg msg, PrintWriter pout) {// always request.
		String request = msg.getValue("Request");
		if (request == null)
			return;
		if (request.equals("ItemIds"))
			onClientRequestItemIds(msg, pout);
		else if (request.equals("ItemInfo"))
			onClientRequestItemInfo(msg, pout);
		else if (request.equals("ItemTopBid"))
			onClientRequestItemTopBid(msg, pout);
		else if (request.equals("ItemBids"))
			onClientRequestItemBids(msg, pout);
		else if (request.equals("PlaceBid"))
			onClientPlaceBid(msg, pout);
		else
			Util.println("unexpected client request: " + request);
	}

	public void onClientRequestItemIds(Msg msg, PrintWriter pout) {
		Msg oMsg = new Msg();
		oMsg.add("Response", "ItemIds");
		Set<Integer> itemIds = getItemIds("");
		for (int i : itemIds)
			oMsg.add("ItemId", Integer.toString(i));
		Util.sendMsg(oMsg, pout);
	}

	public void onClientRequestItemInfo(Msg msg, PrintWriter pout) {
		String itemIdStr = msg.getValue("ItemId");
		if (itemIdStr == null)
			return;
		int itemId = Integer.parseInt(itemIdStr);

		ItemInfo item = getItem(itemId);
		if (item == null)
			return;

		Msg oMsg = new Msg();
		oMsg.add("Response", "ItemInfo");
		oMsg.add("ItemId", itemIdStr);
		oMsg.add("ItemTitle", item.itemTitle);
		oMsg.add("ItemDesc", item.itemDesc);
		oMsg.add("AuctionEndTime", item.auctionEndTime);
		Util.sendMsg(oMsg, pout);
	}

	public void onClientRequestItemTopBid(Msg msg, PrintWriter pout) {
		String itemIdStr = msg.getValue("ItemId");
		if (itemIdStr == null)
			return;
		int itemId = Integer.parseInt(itemIdStr);

		Msg oMsg = new Msg();
		oMsg.add("Response", "ItemTopBid");
		oMsg.add("ItemId", itemIdStr);

		lock.requestCS(itemId, true);
		Bid bid = getTopBid(itemId);
		if (bid != null) {
			oMsg.add("Bid", bid.bid);
			oMsg.add("Bidder", bid.bidder);
			oMsg.add("BidTime", bid.bidTime);
		}
		lock.releaseCS(itemId);
		Util.sendMsg(oMsg, pout);
	}

	public void onClientRequestItemBids(Msg msg, PrintWriter pout) {
		try {
			String itemIdStr = msg.getValue("ItemId");
			if (itemIdStr == null)
				return;
			int itemId = Integer.parseInt(itemIdStr);

			lock.requestCS(itemId, true);
			LinkedList<Bid> bids = getBids(itemId);

			Msg oMsg = new Msg();
			oMsg.add("Response", "ItemBids");
			oMsg.add("ItemId", itemIdStr);
			for (Bid bid : bids) {
				oMsg.add("Bid", Integer.toString(bid.bid));
				oMsg.add("Bidder", bid.bidder);
				oMsg.add("BidTime", bid.bidTime);
			}
			lock.releaseCS(itemId);
			Util.sendMsg(oMsg, pout);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void onClientPlaceBid(Msg msg, PrintWriter pout) {
		String itemIdStr = msg.getValue("ItemId");
		String bidStr = msg.getValue("Bid");
		String bidder = msg.getValue("Bidder");
		String bidTimeStr = msg.getValue("BidTime");
		if (itemIdStr == null || bidStr == null || bidder == null || bidTimeStr == null)
			return;

		int itemId = Integer.parseInt(itemIdStr);
		int bid = Integer.parseInt(bidStr);
		Date bidTime;
		try {
			bidTime = Util.dateFormat.parse(bidTimeStr);
		} catch (ParseException e) {
			e.printStackTrace();
			return;
		}

		lock.requestCS(itemId, false);
		Bid topBid = PlaceBid(itemId, bid, bidder, bidTime);
		Msg oMsg = new Msg();
		oMsg.add("Response", "PlaceBid");
		oMsg.add("ItemId", itemIdStr);
		oMsg.add("Bid", topBid.bid);
		oMsg.add("Bidder", topBid.bidder);
		oMsg.add("BidTime", topBid.bidTime);

		// only if bid changes.
		if (topBid.bid == bid && topBid.bidder.equals(bidder) && topBid.bidTime.equals(bidTime)) // succesful bid.
			publishTopBidUpdate(itemId, topBid);// to other servers.

		lock.releaseCS(itemId);
		Util.sendMsg(oMsg, pout);
	}

	public void publishTopBidUpdate(int itemId, Bid topBid) {
		Msg oMsg = new Msg();
		oMsg.add("Response", "ItemTopBid");
		oMsg.add("TimeStamp", lock.c.tick());
		oMsg.add("ItemId", itemId);
		oMsg.add("Bid", topBid.bid);
		oMsg.add("Bidder", topBid.bidder);
		oMsg.add("BidTime", topBid.bidTime);
		for (PrintWriter pout : peers)
			Util.sendMsg(oMsg, pout);
	}

	public void onServerMsg(Msg msg, PrintWriter pout) {// always
		// contain
		// timeStamp.
		boolean consumed = lock.onMsg(msg, pout);
		if (consumed)
			return;

		String request = msg.getValue("Request");
		String response = msg.getValue("Response");
		if (request != null) {
			if (request.equals("Hello")) { // Hello back to establish initial timestamp.
				onServerRequestHello(msg, pout);
			} else if (request.equals("ItemBids")) {// new node join.
				onServerRequestItemBids(msg, pout);
			}
		} else if (response != null) {
			if (response.equals("ItemTopBid")) {// most cases. other servers
				// push out the change. no
				// request.
				onServerResponseItemTopBid(msg, pout);
			} else if (response.equals("ItemBids")) {// requested info arrived.
				onServerResponseItemBids(msg, pout);
			}
		} else {
			Util.println("unexpected server msg received without request or response tag.");
		}
	}

	public void Hello(PrintWriter pout) {
		Msg oMsg = new Msg();
		oMsg.add("Request", "Hello");
		oMsg.add("TimeStamp", lock.c.getValue());
		Util.sendMsg(oMsg, pout);
	}

	public void onServerRequestHello(Msg msg, PrintWriter pout) {
		Msg oMsg = new Msg();
		oMsg.add("Response", "Hello");
		oMsg.add("TimeStamp", lock.c.getValue());
		Util.sendMsg(oMsg, pout);
	}

	public void onServerRequestItemBids(Msg msg, PrintWriter pout) {
		try {
			String itemIdStr = msg.getValue("ItemId");
			if (itemIdStr == null)
				return;
			int itemId = Integer.parseInt(itemIdStr);
			LinkedList<Bid> bids = getBids(itemId);

			Msg oMsg = new Msg();
			oMsg.add("Response", "ItemBids");
			oMsg.add("TimeStamp", lock.c.getValue());
			oMsg.add("ItemId", itemId);

			if( bids != null){
				for (Bid bid : bids) {
					oMsg.add("Bid", Integer.toString(bid.bid));
					oMsg.add("Bidder", bid.bidder);
					oMsg.add("BidTime", bid.bidTime);
				}
			}
			Util.sendMsg(oMsg, pout);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void onServerResponseItemBids(Msg msg, PrintWriter pout) {
		// to replace my bids. happen only during initialization.
		// requester in CS().
		initialItemBidsInProgress = false; // received the msg.
		try {
			String itemIdStr = msg.getValue("ItemId");
			if (itemIdStr == null)
				return;
			int itemId = Integer.parseInt(itemIdStr);
			LinkedList<Bid> bids = getBids(itemId);
			if( bids != null){
				bids.clear();
	
				// each bid starts with bid and ends with bidTime. add bid in when
				// seeing bidTime.
				int bid = 0;
				String bidder = "";
				Date bidTime;
				for (TaggedValue tv : msg.data) {
					if (tv.tag.equals("Bid"))
						bid = Integer.parseInt(tv.value);
					else if (tv.tag.equals("Bidder"))
						bidder = tv.value;
					else if (tv.tag.equals("BidTime")) {
						bidTime = Util.dateFormat.parse(tv.value);
						bids.addLast(new Bid(bid, bidder, bidTime));
					}
				}
			}
		} catch (ParseException e) {
			e.printStackTrace();
		} finally {
			Util.notify(this);
		}
	}

	public void onServerResponseItemTopBid(Msg msg, PrintWriter pout) {// no
		// request,only response for sync requester in CS().
		try {
			String itemIdStr = msg.getValue("ItemId");
			String bidStr = msg.getValue("Bid");
			String bidder = msg.getValue("Bidder");
			String bidTimeStr = msg.getValue("BidTime");
			if (itemIdStr == null || bidStr == null || bidder == null || bidTimeStr == null)
				return;
			int itemId = Integer.parseInt(itemIdStr);
			int bid = Integer.parseInt(bidStr);
			Date bidTime = Util.dateFormat.parse(bidTimeStr);
			PlaceBid(itemId, bid, bidder, bidTime);
		} catch (ParseException e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		if (args.length < 1) {
			Util.println("invalid parameter. usage:\n java BidServer <ID>");
			return;
		}
		BidServer server = new BidServer();
		if (!server.init(Integer.parseInt(args[0]), "/tmp/bidServer.conf")) {
			Util.println("initialization failed.");
			return;
		}
		Util.println("bidServer " + server.lock.srcId + " started successfully");
		server.HouseKeep();
	}
}
